package indi.zhaosheng.rocktmq.consumer;

import indi.zhaosheng.rocktmq.exception.RocketmqRuntimeException;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;

/**
 * @author muluo
 * @description
 * @date 2020/5/5 15:11
 */
public abstract class BaseConsumer {

    /**
     * rocketMq地址
     */
    protected String namesrvAddress;
    protected String topic;
    protected String consumerGroup;
    protected String tag;
    protected ConsumerAction consumerAction;

    protected BaseConsumer(String topic, String consumerGroup) {
        this.topic = topic;
        this.consumerGroup = consumerGroup;
    }

    /**
     * <p>
     * 初始化namesrvAddress，调用{@link BaseConsumer#setNamesrvAddress(String)}方法；
     * 初始化consumerAction，调用{@link BaseConsumer#setConsumerAction(String, ConsumerAction)}方法
     * </p>
     *
     * @return void
     * @author muluo
     * @date 2020/5/5 15:27
     */
    protected abstract void init();

    protected void setNamesrvAddress(String namesrvAddress) {
        this.namesrvAddress = namesrvAddress;
    }

    protected void setConsumerAction(String tag, ConsumerAction consumerAction) {
        this.tag = tag;
        this.consumerAction = consumerAction;
    }

    /**
     * <p>
     * 启动consumer
     * </p>
     *
     * @return void
     * @author muluo
     * @date 2020/5/11 21:50
     */
    public abstract void start() throws MQClientException;

    protected void check() {
        if (null == this.namesrvAddress) {
            throw RocketmqRuntimeException.build("请初始化namesrvAddress");
        }
        if (null == this.consumerAction) {
            throw RocketmqRuntimeException.build("请初始化consumerAction");
        }
    }

    protected DefaultMQPushConsumer initConsumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(namesrvAddress);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe(topic, tag);
        return consumer;
    }
}
